package ex.datastream.windows;

import ex.datastream.functions.sourceFunction.SourceFuncByEvent;
import ex.vo.Event;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.delta.DeltaFunction;
import org.apache.flink.streaming.api.windowing.assigners.GlobalWindows;
import org.apache.flink.streaming.api.windowing.evictors.DeltaEvictor;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.triggers.ContinuousProcessingTimeTrigger;
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;
import org.apache.flink.util.Collector;


/**
 * ref:https://blog.csdn.net/wangpei1949/article/details/102996170
 */
public class DeltaEvictorTest {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.addSource(new SourceFuncByEvent())
                .keyBy(event -> 1)
                .windowAll(GlobalWindows.create())
                .trigger(ContinuousProcessingTimeTrigger.of(Time.seconds(4)))
                 // 剔除与最后一个元素值差大于50的元素
                .evictor(DeltaEvictor.of(50.0, new DeltaFunction<Event>() {
                    @Override
                    public double getDelta(Event event, Event data1) {
                        if(event.key-data1.key>50){
                            System.out.printf("oldEey:%s,newKey:%s,diff:%s\n",event.key,data1.key,event.key-data1.key);
                        }
                        return event.key-data1.key;
                    }
                },true))
                .process(
                        new ProcessAllWindowFunction<Event, String, GlobalWindow>() {
                            @Override
                            public void process(Context context, Iterable<Event> elements, Collector<String> out) throws Exception {
                                out.collect("窗口中有" + elements.spliterator().getExactSizeIfKnown() + "条元素");
                                elements.forEach(a-> System.out.printf(a.key+","));
                                System.out.println("\n*******************************************************");
                            }
                        }
                )
                .print();
        env.execute();
    }
//窗口中有12条元素
//23,73,50,28,28,60,22,13,77,52,75,30,
//*******************************************************
//窗口中有32条元素
//23,73,50,28,28,60,22,13,77,52,75,30,79,69,3,7,67,86,51,96,62,48,98,66,29,99,31,32,20,11,71,23,
//*******************************************************
//oldEey:77,newKey:23,diff:54
//oldEey:75,newKey:23,diff:52
//oldEey:79,newKey:23,diff:56
//oldEey:86,newKey:23,diff:63
//oldEey:96,newKey:23,diff:73
//oldEey:98,newKey:23,diff:75
//oldEey:99,newKey:23,diff:76
//窗口中有43条元素
//23,50,28,28,60,22,13,52,30,69,3,7,67,51,62,48,66,29,31,32,20,11,71,23,84,34,82,42,70,3,23,44,66,58,77,41,62,7,4,21,5,66,63,
}

